package com.googlecode.jvmpart.proc.jdbc;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.googlecode.jvmpart.proc.JvmProcessException;

/**
 * @author chunzhan.he
 *
 * @param <T>
 */
public abstract class ChunkProcessor<T> extends SimpleRowProcessor{
	private static final Log logger = LogFactory.getLog(ChunkProcessor.class);
	private long commited = 0;
	private List<T> chunk = new ArrayList<T>();
	private boolean commit = false;
	@Override
	public void rowProcess(ResultSet rs, long row) throws JvmProcessException {
		commit = (row%getCommitInterval()==0);
		T t = null;
		try {
			t = process(mapRow(rs, row));
		} catch(SQLException e) {
			throw new JvmProcessException("mapRow error",e);
		}
		if(t==null) return;
		chunk.add(t);
		if(commit) {
			try {
				doCommit();
			} catch (JvmProcessException e) {
				logger.error(e, e);
				throw e;
			} finally {
			}
		}		
	}

	@Override
	public void allProcessed(long processed, long error) throws JvmProcessException {
		if(!chunk.isEmpty()) {
			doCommit();
		}
		allChunkProcessed(processed, commited);
	}
	private void doCommit() throws JvmProcessException {
		try {
			write(chunk);
			getConn().commit();
			commited++;
			chunk.clear();
		} catch (SQLException e) {
			logger.error(e, e);
			try {
				getConn().rollback();
			} catch (SQLException e1) {}
			throw new JvmProcessException("commit error", e);
		} finally {
		}		
	}

	@Override
	public boolean whenError(ResultSet rs, long row, Throwable t) {
		if(commit) {
			boolean breakFlag = commitChunkError(chunk, t);
			chunk.clear();
			return breakFlag;
		} else return super.whenError(rs, row, t);
	}
	/**
	 * @param processed total of processed
	 * @param commited number of commited count
	 */
	public abstract void allChunkProcessed(long processed, long commited);
	/**
	 * @param chunk current processing chunk
	 * @param t processed error exception
	 * @return if true will break the whole processing,false will continue
	 */
	public abstract boolean commitChunkError(List<T> chunk, Throwable t);
	public abstract T mapRow(ResultSet rs, long row) throws SQLException;
	public abstract T process(T t) throws JvmProcessException;
	public abstract void write(List<T> list) throws JvmProcessException;
	public abstract int getCommitInterval();

}
